package com.ckm.listener;

import com.mongodb.client.ChangeStreamIterable;
import com.mongodb.client.model.Aggregates;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.client.model.changestream.FullDocument;
import lombok.extern.slf4j.Slf4j;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.data.mongodb.MongoDatabaseFactory;
import org.springframework.data.mongodb.core.ChangeStreamEvent;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.mapping.event.AbstractMongoEventListener;
import org.springframework.data.mongodb.core.mapping.event.AfterSaveEvent;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer;
import org.springframework.data.mongodb.core.query.Criteria;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

@Slf4j
public class DynamicMongeSourceListener{

    public DynamicMongeSourceListener(MongoDatabaseFactory databaseFactory){
        DefaultMessageListenerContainer container = new DefaultMessageListenerContainer(new MongoTemplate(databaseFactory));

//        ChangeStreamRequest build = ChangeStreamRequest.builder()
//                .collection("hello")
//                .filter(Aggregation.newAggregation
//                        (Aggregation.match
//                                (Criteria.where("operationType")
//                                        .in("insert", "delete", "update", "replace"))))
//                .publishTo(handleChange())
//                .build();
        MongoTemplate mongoTemplate = new MongoTemplate(databaseFactory);
        Bson filter = Filters.in("operationType", "insert", "delete", "update", "replace");
        List<Bson> bson = Arrays.asList(Aggregates.match(filter));
        ChangeStreamIterable<Document> changeStreamDocuments =
                mongoTemplate.getCollection("hello")
                        .watch(bson)
                        .fullDocument(FullDocument.UPDATE_LOOKUP);
        for (ChangeStreamDocument<Document> changeStreamDocument : changeStreamDocuments) {
            Document document = changeStreamDocument.getFullDocument();
            System.out.println(document);
        }
    }
}
